package com.wondersgroup.core.util;

import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 线程队列处理线程, 内置一个有序队列, 通过调用push方法将要处理的线程加入队列中等待处理<br/>
 * 该类提供的方法是线程安全的, 创建本类实例的时候建议使用单例模式
 * @author wanglei
 */
public class QueueProcessThread extends Thread{
    protected static final Logger logger = LoggerFactory.getLogger(QueueProcessThread.class);
    
    //线程运行标识
    protected boolean running = false;
    //发送状态是否需要关闭
    protected boolean closed = false;
    //是否继续接收新的处理线程
    protected boolean acceptPush = true;
    protected final ReentrantLock takeLock = new ReentrantLock();
    protected final ReentrantLock putLock = new ReentrantLock();
    protected final Condition notEmpty = takeLock.newCondition();
    //发送队列
    protected final Queue<Thread> queue = new LinkedList<>();
    //线程池默认大小
    protected static final int THREAD_POOL_SIZE = 32;
    //固定大小的线程池，限制同时发送的线程数量，避免无限增长降低效率
    protected ExecutorService threadPool = Executors.newFixedThreadPool(THREAD_POOL_SIZE);
    
    protected void signalNotEmpty() {
        final ReentrantLock lock = this.takeLock;
        lock.lock();
        try {
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
    }
    
    protected void fullyLock() {
        putLock.lock();
        takeLock.lock();
    }
    
    protected void fullyUnlock() {
        takeLock.unlock();
        putLock.unlock();
    }
    
    /**
     * 加入发送队列
     * @param thread
     */
    public void push(Thread thread){
        final ReentrantLock lock = this.putLock;
        lock.lock();
        try {
            if (acceptPush && thread!=null) {
                queue.offer(thread);
                if (!running) {
                    running = true;
                    start();
                }
                //加入队列后通知发送线程解除等待
                this.signalNotEmpty();
            }
        } catch (Exception ex) {
            logger.info(null, ex);
            throw ex;
        } finally {
            lock.unlock();
        }
    }
    
    @Override
    public void run() {
        final ReentrantLock lock = this.takeLock;
        lock.lock();
        try {
            while (running) {
                Thread thread = queue.poll();
                if (thread == null) {
                    if (!closed) {
                        //如果发送队列为空同时没有将线程标识为close则进入等待
                        notEmpty.await();
                    } else {
                        //否则关闭发送线程
                        running = false;
                        if(!threadPool.isShutdown()){
                            threadPool.shutdown();
                        }
                    }
                } else {
                    //将线程加入线程池
                    if(!threadPool.isShutdown()){
                        threadPool.execute(thread);
                    }
                }
            }
        } catch (Exception ex) {
            logger.info(null, ex);
        } finally {
            lock.unlock();
        }
    }
    
    /**
     * 重置状态，使其可以继续接收push操作和消息发送
     */
    public void reset() {
        this.fullyLock();
        try {
            this.closed = false;
            this.acceptPush = true;
            if(!this.threadPool.isShutdown()){
                this.threadPool.shutdown();
            }
            this.threadPool = Executors.newFixedThreadPool(THREAD_POOL_SIZE);
        } finally {
            this.fullyUnlock();
        }
    }

    /**
     * 使push操作不再可用, 发送线程将在线程池等待队列中的所有任务发送完成后再关闭
     */
    public void close() {
        this.fullyLock();
        try {
            this.closed = true;
            this.acceptPush = false;
            //如果发送线程在等待则解除等待
            this.signalNotEmpty();
        } finally {
            this.fullyUnlock();
        }
    }
    
    /**
     * 尝试立即关闭发送线程，使push操作不再可用
     */
    public void shutdown() {
        this.fullyLock();
        try {
            this.running = false;
            this.closed = true;
            this.acceptPush = false;
            this.queue.clear();
            if(!this.threadPool.isShutdown()){
                this.threadPool.shutdownNow();
            }
            //如果发送线程在等待则解除等待
            this.signalNotEmpty();
        } finally {
            this.fullyUnlock();
        }
    }
}
